package com.mutouren.modules.datapush.server;

import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;

import com.mutouren.common.log.LogAlias;
import com.mutouren.common.log.LogManager;
import com.mutouren.common.log.Logger;
import com.mutouren.common.thread.ThreadPoolManager;
import com.mutouren.common.utils.FileUtils;
import com.mutouren.modules.datapush.base.DataPushLogAlias;
import com.mutouren.modules.datapush.base.GradeQueue;
import com.mutouren.modules.datapush.base.OutOfPermitBoundsException;
import com.mutouren.modules.datapush.base.PushMessage;
import com.mutouren.modules.datapush.base.PushServer;
import com.mutouren.modules.datapush.storage.StorageHelper;
import com.mutouren.modules.datapush.strategy.ConcurrentStrategy;
import com.mutouren.modules.datapush.strategy.ConcurrentStrategyManager;
import com.mutouren.modules.datapush.strategy.RepeatStrategy;
import com.mutouren.modules.datapush.strategy.RepeatStrategyManager;

public abstract class AbstractServer implements PushServer{

	private String serverAlias;
	private GradeQueue<PushMessage> gradeQueue;
	private Queue<PushMessage> waitingQueue;
	private Map<String, PushMessage> processingMap;
	
	private ExecutorService leader;
	private ExecutorService workers; 
	
	private Object waitTaskEvent = new Object();
	private Object waitWorkerEvent = new Object();
	private Object lockQueue = new Object();
	
	private volatile int curDispatcherCount = 0;
	protected int maxPermitSize = 1000000;
	
	private volatile String storageFilePath = "";
	private volatile boolean isOpened = false;
		
	protected ConcurrentStrategyManager concurrentStrategyManager = new ConcurrentStrategyManager();
	protected RepeatStrategyManager repeatStrategyManager = new RepeatStrategyManager();
		
	protected final static Logger errorLogger = LogManager.getLogger(LogAlias.SystemError.name());
	protected final static Logger runLogger = LogManager.getLogger(LogAlias.SystemRun.name());
	protected final static Logger monitorLogger = LogManager.getLogger(DataPushLogAlias.DataPushMonitor.name());
	protected final static Logger sendLogger = LogManager.getLogger(DataPushLogAlias.DataPushSend.name());

	@Override	
	public String getServerAlias() {
		return this.serverAlias;
	}
	
	public AbstractServer(String serverAlias, int maxThreadPoolSize) {
		if (StringUtils.isBlank(serverAlias)) {
			throw new RuntimeException("serverAlias name is blank");
		}		
		this.serverAlias = serverAlias;
		
		gradeQueue = new GradeQueue<PushMessage>(3, 10);
		waitingQueue = new LinkedList<PushMessage>();
		processingMap = new HashMap<String, PushMessage>();
		
		leader = ThreadPoolManager.createSingleThreadExecutor();
		workers = ThreadPoolManager.createThreadPool(0, maxThreadPoolSize, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
	}
	
//	@Override
//	public boolean isOpen() {
//		return this.isOpened;
//	}
	
	private void run() {
		this.isOpened = true;
		leader.execute(()-> this.dispatcherTask());	
	}
	
	private void stop() {
		this.isOpened = false;
		leader.shutdownNow();
		workers.shutdownNow();
		
		try {
			leader.awaitTermination(10, TimeUnit.SECONDS);
			workers.awaitTermination(10, TimeUnit.SECONDS);			
		} catch (Exception e) {
			errorLogger.error(this.getServerAlias() + " server awaitTermination error", e);
		}
	}
	
	private void load() {
		if (StringUtils.isBlank(storageFilePath)) return;
		if (!FileUtils.isExist(storageFilePath)) return;
		
		synchronized(this.lockQueue) {	
			StorageHelper.load(storageFilePath, gradeQueue);
			FileUtils.deleteFile(storageFilePath);
			runLogger.info(this.getServerAlias() + " server load message count: " + gradeQueue.getSize());	
		}
		
	}
	
	private void save() {
		if (StringUtils.isBlank(storageFilePath)) return;
		if (FileUtils.isExist(storageFilePath)) {
			FileUtils.deleteFile(storageFilePath);
		}
		
		synchronized(this.lockQueue) {	
			int saveCount = waitingQueue.size() + gradeQueue.getSize();
			StorageHelper.save(storageFilePath, waitingQueue);
			StorageHelper.save(storageFilePath, gradeQueue);
				
			// processing don't consider
				
			runLogger.info(this.getServerAlias() + " server save message count: " + saveCount);
		}
	}
	
	private long lastMoniterTime = 0;
	private void debugMoniter() {
		long curTime = new Date().getTime();
		
		if ((curTime - lastMoniterTime) > 5000) {
			String log = String.format("serverAlias:%s,gradeQueue:%d,waitingQueue:%d,processingMap:%d,workingCount:%d,curDispatcherCount:%d", 
					serverAlias, gradeQueue.getSize(), waitingQueue.size(), processingMap.size(), 
					concurrentStrategyManager.getConcurrentTotalCountForDebug(), curDispatcherCount);
			monitorLogger.info(log);
			lastMoniterTime = curTime;
		}
	}
	
	private void dispatcherTask() {
		do {
			debugMoniter();
			PushMessage message = pollMessageFromQueue();
			processMessage(message);
		
			if ((getGradeQueueSize(0) > 0)&&(this.isOpened)) continue;		
							
			synchronized(this.waitTaskEvent) {
				try {
					if (waitingQueue.size() == 0) {
						waitTaskEvent.wait(100);
					} else {
						waitTaskEvent.wait(5);
					}
				} catch (InterruptedException e) {
					//?errorLogger.warn(this.getServerAlias() + " dispatcherTask Interrupted", e);
				}
			}
			
		} while(ThreadPoolManager.isRun()&&this.isOpened);		
	}
	
	private void processMessage(PushMessage message) {
		if (message == null) return;
		
		if (isExpired(message)) {
			refundMessageToQueue(message, false, true);
			return;
		}					
		if (!isPermitSend(message)) {
			refundMessageToQueue(message, false, false);
			return;
		}				
		
		if (concurrentStrategyManager.isConcurrent(message)) {
			refundMessageToQueue(message, true, false);
			return;
		}		
		
		MyTask myTask = new MyTask(this.waitWorkerEvent, new Runnable() {
			@Override
			public void run() {
				try {
					concurrentStrategyManager.addConcurrentCount(message.getMethodAlias());					
					send(message);					
				} catch(Throwable t) {
					errorLogger.error(AbstractServer.this.getServerAlias() + " server.processMessage.run error", t);
				} finally {
					afterSend(message);
					concurrentStrategyManager.subtractConcurrentCount(message.getMethodAlias());
				}
			}
		});	
		
		do {
			try {
				this.workers.submit(myTask);				
				return;
			} catch(RejectedExecutionException e) {
				
	            synchronized(this.waitWorkerEvent) {
	            	try {
						this.waitWorkerEvent.wait(1000);
					} catch (InterruptedException e1) {
						errorLogger.error(this.getServerAlias() + " server.processMessage wait error", e1);
					}
	            }				
			}
			
			// bug: if isOpened=false, cause loss task.
			if (this.isOpened == false) {
				refundMessageToQueue(message, false, false);
			}
			
		} while(ThreadPoolManager.isRun()&&this.isOpened);			
	}
	
	protected abstract void send(PushMessage message);
	
	private void afterSend(PushMessage message) {	
		boolean isRepeat = isPermitRepeat(message);		
			
		synchronized(this.lockQueue) {
			PushMessage obj = processingMap.remove(message.getGuid());
			if (obj == null) {
				RuntimeException e = new RuntimeException(this.getServerAlias() + " logic fatal error: processingMap not find message");
				errorLogger.error("",e);
				throw e;
			}
			
			if (message.getRequestCount() == 0) {
				isRepeat = true; // for no http etc error
				waitingQueue.add(message);
			} else {
				if (isRepeat) {
					gradeQueue.add(message);
				}
			}
		}
		
		if (!isRepeat) {
			onFinish(message);
		}
	}
	
	protected void onFinish(PushMessage message) {
		//runLogger.info("onFinish: " + message.getId());
	}
	
	private PushMessage pollMessageFromQueue() {			
		synchronized(this.lockQueue) {
			PushMessage result = null;			
			curDispatcherCount++;
			
			if(curDispatcherCount % 3 == 0) {
				result = waitingQueue.poll();
				if (result != null) {
					processingMap.put(result.getGuid(), result);
					return result;
				}
			}
			
			result = gradeQueue.poll();
			if (result != null) {
				processingMap.put(result.getGuid(), result);
			}
			return result;
		}
	}
	
	private void refundMessageToQueue(PushMessage message, boolean isWaiting, boolean isExpired) {
		synchronized (this.lockQueue) {
			PushMessage obj = processingMap.remove(message.getGuid());
			if (obj == null) {
				RuntimeException e = new RuntimeException(this.getServerAlias() + " logic fatal error: processingMap not find message");
				errorLogger.error("",e);
				throw e;				
			}
			
			if (isExpired) {
				onFinish(obj);
				return;
			}
			
			if (isWaiting) {
				waitingQueue.add(obj);
			} else {
				gradeQueue.add(obj);
			}
		}
	}	

	@Override
	public void receive(PushMessage message) {
		
		if (this.isOpened == false) {
			throw new RuntimeException(this.getServerAlias() + " server is close, should open");
		}
				
		synchronized(this.lockQueue) {
			if (gradeQueue.getSize() > maxPermitSize) {
				OutOfPermitBoundsException e = new OutOfPermitBoundsException(this.getServerAlias() + " server maxPermitSize is " + maxPermitSize);
				errorLogger.error("",e);
				throw e;				
			}
			gradeQueue.add(message);
		}
		synchronized(this.waitTaskEvent) {
			this.waitTaskEvent.notify();
		}
	}
	
	private int getGradeQueueSize(int grade) {
		synchronized(this.lockQueue) {
			return gradeQueue.getSize(grade);
		}
	}
	
	protected int getQueueSize() {
		synchronized(this.lockQueue) {
			return gradeQueue.getSize() + this.waitingQueue.size() + this.processingMap.size();
		}
	}

	@Override
	public void open(String filePath) {
		synchronized(this) {
			if (isOpened) {
				runLogger.info(this.getServerAlias() + " server already opened");
				return;
			}
			
			this.storageFilePath = filePath;
			this.load();
			this.run();			
		}
	}

	@Override
	public void close() {
		synchronized(this) {
			if (!isOpened) {
				runLogger.info(this.getServerAlias() +  " server already closed");
				return;
			}
			
			this.stop();
			this.save();			
		}
	}
	
	protected boolean isExpired(PushMessage message) {
		return repeatStrategyManager.isExpired(message.getMethodAlias(), message);
	}
	protected boolean isPermitSend(PushMessage message) {
		return repeatStrategyManager.isPermitSend(message.getMethodAlias(), message);
	}
	protected boolean checkSuccessFlag(String methodAlias, String value) {
		return repeatStrategyManager.checkSuccessFlag(methodAlias, value);
	}
	protected boolean isPermitRepeat(PushMessage message) {
		return repeatStrategyManager.isPermitRepeat(message.getMethodAlias(), message);
	}
	
	@Override
	public void addConcurrentStrategy(ConcurrentStrategy strategy) {
		if (isOpened) {
			throw new RuntimeException(this.getServerAlias() + " server is opened, can't addConcurrentStrategy");
		}	
		concurrentStrategyManager.addConcurrentStrategy(strategy);
	}

	@Override
	public void addRepeatStrategy(RepeatStrategy strategy) {
		if (isOpened) {
			throw new RuntimeException(this.getServerAlias() + " server is opened, can't addRepeatStrategy");
		}			
		repeatStrategyManager.addRepeatStrategy(strategy);
	}
	
    public static class MyTask implements Runnable {
        private Object lock;
        private Runnable task;

        public MyTask(Object lock, Runnable task) {
            this.lock = lock;
            this.task = task;
        }

        @Override
        public void run() {
            task.run();
            synchronized(lock) {
            	lock.notify();
            }
        }
    }		
}
